//////////////////////////////////////////////////////////////////////////
///	\file:	SocketReader.cpp
///	Copyright (MingYou) 2009 - All Rights Reserved
///	Author:	Robert.Ma
///	Date:	[2009/05/14]
///	Description:
///		
//////////////////////////////////////////////////////////////////////////
#include "StdAfx.h"
#include "SocketReader.h"
#include "InlineFunction.h"
#include "CETool.h"



CSocketReader::CSocketReader(void)
		: m_iCurBufferSize(0),
		m_pSocket(NULL),
		m_pRecBuf(NULL),
		m_iSocketError(0),
		m_hBufMutex(NULL),
		m_dwParentID(ULONG_MAX)
{}

CSocketReader::~CSocketReader(void)
{
	Release();
}

BOOL CSocketReader::Initialize(SOCKET* pSocket, DWORD dwParentID)
{
	Release();

	m_pSocket		= pSocket;
	m_iSocketError	= 0;
	m_dwParentID	= dwParentID;
	m_pRecBuf		= new char[TCPBUFFERSIZE];

	return (NULL != CMThread::Create());
}

void CSocketReader::Release()
{
	//	Close thread firstly, or maybe cause an error.
	CMThread::Close();

	if ( LockBuffer() )
	{
		m_iCurBufferSize = 0;

		UnlockBuffer();
	}

	::ClearArrayPointer(m_pRecBuf);
	::ErasePoiter(m_pSocket);

	m_iSocketError = 0;
}

int CSocketReader::GetBufferSize()
{
	if ( LockBuffer() )
	{
		int nSize = m_iCurBufferSize;
		UnlockBuffer();
		return nSize;
	}
	else
	{
		return -1;
	}
}

DWORD CSocketReader::Execute()
{
	MSG msg;

	while ( ::GetMessage(&msg, NULL, 0, 0) )
	{
		switch ( msg.message )
		{
		case TM_NR_READ_START:
			ReadSocketData();
			ActiveReadAction(READ_SLEEP_TIME);
			break;

		case TM_THREAD_EXIT:
			//	Do some clear...
			//MTRACE(L"CSocketReader thread exit.");
			return 0;

		default:
			ASSERT(FALSE);
			break;
		}
	}

	return 0;
}

void CSocketReader::PreClose()
{
	SafePostMessage(GetThreadID(), TM_THREAD_EXIT);
}

void CSocketReader::PreCreate()
{
	if ( NULL == m_hBufMutex )
	{
		m_hBufMutex = ::CreateMutex(NULL, FALSE, L"Read buffer Mutex");
	}
}

void CSocketReader::AfterCreate()
{
	ActiveReadAction(0);
}

void CSocketReader::AfterClose()
{
	if ( NULL != m_hBufMutex )
	{
		::CloseHandle(m_hBufMutex);
		m_hBufMutex = NULL;
	}
}

int CSocketReader::ReadSocketData()
{
	int iRecSize = -1;

	if ( 0 == m_iSocketError )
	{
		iRecSize = ::recv(*m_pSocket, m_pRecBuf, TCPBUFFERSIZE, 0);

		if ( SOCKET_ERROR == iRecSize )
		{
			m_iSocketError = ::WSAGetLastError();

			SafePostMessage(m_dwParentID, TM_SOCKET_ERROR, m_iSocketError);

			return -1;
		}

		if ( 0 < iRecSize )
		{
			ASSERT( MAX_BUFFERSIZE > m_iCurBufferSize + iRecSize );
			LockBuffer();
			::memcpy_s(m_byCacheBuffer + m_iCurBufferSize, MAX_BUFFERSIZE - m_iCurBufferSize, m_pRecBuf, iRecSize);
			m_iCurBufferSize += iRecSize;
			UnlockBuffer();
			//////////////////////////////////////////////////////////////////////////
			//	Check if there is a package in buffer, if has a package, post a message
			//	to main thread and reset buffer.
			SafePostMessage(m_dwParentID, TM_NR_READ_RECV);
		}
	}

	return iRecSize;
}

BOOL CSocketReader::LockBuffer()
{
	return (WAIT_OBJECT_0 == ::WaitForSingleObject(m_hBufMutex, WAITBUFFER_TIME));
}

void CSocketReader::UnlockBuffer()
{
	::ReleaseMutex(m_hBufMutex);
}

int CSocketReader::CopyBuffer(BYTE* pbyGet, int iLen)
{
	if ( LockBuffer() )
	{
		if ( iLen >= m_iCurBufferSize )
		{
			::memcpy_s(pbyGet, iLen, m_byCacheBuffer, m_iCurBufferSize);
			m_iCurBufferSize = 0;
		}
		else
		{
			::memcpy_s(pbyGet, iLen, m_byCacheBuffer, iLen);
			m_iCurBufferSize -= iLen;
			::memcpy_s(m_byShadowBuffer, m_iCurBufferSize, m_byCacheBuffer, m_iCurBufferSize);
			::memcpy_s(m_byCacheBuffer, m_iCurBufferSize, m_byShadowBuffer, m_iCurBufferSize);
		}
		UnlockBuffer();

		return m_iCurBufferSize;
	}
	else
	{
		return -1;
	}
}

void CSocketReader::ActiveReadAction(DWORD dwDelay)
{
	::Sleep(dwDelay);
	SafePostMessage(GetThreadID(), TM_NR_READ_START);
}
